package com.shujia.hbase.api;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class Demo3HbaseApi {

    private HConnection connection;
    private HBaseAdmin hBaseAdmin;

    @Before
    public void init() {

        //创建hadoop配置文件对象
        Configuration configuration = new Configuration();

        //指定zk地址
        configuration.set("hbase.zookeeper.quorum", "node1:2181,node2:2181,node3:2181");


        try {
            //建立连接   和zk的连接
            connection = HConnectionManager.createConnection(configuration);

            hBaseAdmin = new HBaseAdmin(configuration);

            System.out.println("连接建立成功");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @Test
    public void createTable() {
        HTableDescriptor student = new HTableDescriptor("student");
        HColumnDescriptor info = new HColumnDescriptor("info");
        //info.setMaxVersions(10);
        //info.setTimeToLive(100);

        //增加列簇
        student.addFamily(info);

        try {
            //创建表
            hBaseAdmin.createTable(student);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    @Test
    public void dropTable() {

        //删除表
        try {
            if (hBaseAdmin.tableExists("student")) {
                hBaseAdmin.disableTable("student");
                hBaseAdmin.deleteTable("student");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @Test
    public void put() {
        //插入单行数据


        try {
            //获取表的实例
            HTableInterface student = connection.getTable("student");


            //创建put对象
            Put put = new Put("001".getBytes());

            //增加列
            put.add("info".getBytes(), "name".getBytes(), "zhsngsan".getBytes());

            //插入数据
            student.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void putAll() {

        //批量插入数据
        ArrayList<Put> puts = new ArrayList<Put>();

        try {

            //获取表的实例
            HTableInterface student = connection.getTable("student");


            BufferedReader bufferedReader = new BufferedReader(new FileReader("data/students.txt"));

            String line;
            while ((line = bufferedReader.readLine()) != null) {

                String[] split = line.split(",");

                String id = split[0];
                String name = split[1];
                Integer age = Integer.parseInt(split[2]);
                String gender = split[3];
                String clazz = split[4];


                //构建put对象   一学号作为rowkey
                Put put = new Put(id.getBytes());

                //增加列
                put.add("info".getBytes(), "name".getBytes(), name.getBytes("utf-8"));

                //Bytes.toBytes(age)  将基本数据类型转换成字节数组
                put.add("info".getBytes(), "age".getBytes(), Bytes.toBytes(age));

                put.add("info".getBytes(), "gender".getBytes(), gender.getBytes());
                put.add("info".getBytes(), "clazz".getBytes(), clazz.getBytes());

                //将put放到集合中
                puts.add(put);

            }


            //批量插入数据
            student.put(puts);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Test
    public void get() {

        //获取表的实例
        try {
            HTableInterface student = connection.getTable("student");

            //构建get对象   指定rowkey
            Get get = new Get("1500100999".getBytes());

            //执行查询  返回 result
            Result result = student.get(get);

            //取出rowkey
            String id = Bytes.toString(result.getRow());


            // 1、直接通过列名取值   如果列名不存在 返回对应类型的默认值
            byte[] bytes = result.getValue("info".getBytes(), "name".getBytes());
            //通过String的构造函数 构建字符串
            //String name = new String(bytes, "utf-8");

            //hbase  推荐方式
            String name = Bytes.toString(bytes);

            //int  类型  Bytes.toInt  转换
            int age = Bytes.toInt(result.getValue("info".getBytes(), "age".getBytes()));

            String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));

            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz);


            System.out.println("==============");

            // 2、获取所有单元格
            List<Cell> cells = result.listCells();

            //每一个cell 就是一个单元格的数据
            for (Cell cell : cells) {

                //列名
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));

                if ("age".equals(qualifier)) {
                    //取出列值
                    byte[] value = CellUtil.cloneValue(cell);
                    int i = Bytes.toInt(value);
                    System.out.println(i);
                } else {
                    byte[] value = CellUtil.cloneValue(cell);
                    String s = Bytes.toString(value);
                    System.out.println(s);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @Test
    public void scan() {
        try {
            HTableInterface student = connection.getTable("student");

            Scan scan = new Scan();
            //指定开始key和结束key   加了这个条件之后就不会进行全表扫描了
            scan.setStartRow("1500100986".getBytes());
            scan.setStopRow("1500100996".getBytes());

            //可以指定扫描部分列簇或者列
            scan.addColumn("info".getBytes(), "name".getBytes());
            scan.addColumn("info".getBytes(), "age".getBytes());

            //扫描全表
            ResultScanner scanner = student.getScanner(scan);

            //遍历返回的迭代器   每一个Result  代表一个rowkey的数据
            Result next;
            while ((next = scanner.next()) != null) {
                //打印结果
                print(next);
            }


        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @Test
    public void scanSex() {

        //获取所有性别为男的学生

        try {
            HTableInterface student = connection.getTable("student");
            Scan scan = new Scan();


            //创建比较器    正则比较器
            RegexStringComparator comparator = new RegexStringComparator("文科.");

            //包含字符串比较器
            SubstringComparator comparator1 = new SubstringComparator("文科");

            //前缀比较器
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("文科".getBytes());

            //二进制比较器
            BinaryComparator binaryComparator = new BinaryComparator("文科一班".getBytes());

            //创建过滤器   列值过滤器   需要进行全表扫描
            SingleColumnValueFilter valueFilter = new SingleColumnValueFilter("info".getBytes(), "clazz".getBytes(),
                    CompareFilter.CompareOp.EQUAL, binaryComparator);

            //增加过滤器
            scan.setFilter(valueFilter);

            ResultScanner scanner = student.getScanner(scan);

            Result result;
            while ((result = scanner.next()) != null) {
                String sex = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));

                //在客户端进行过滤   还是会进行全表扫描
                //if ("男".equals(sex)) {
                print(result);
                //}
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @Test
    public void filterRowkey() {

        //取出学号  以 15001001  开头的学生

        try {
            HTableInterface student = connection.getTable("student");

            //前缀比较器
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("15001001".getBytes());

            //行建过滤器
            RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);


            //二进制比较器
            BinaryComparator binaryComparator = new BinaryComparator("女".getBytes());

            //取出性别为女的学生
            SingleColumnValueFilter valueFilter = new SingleColumnValueFilter("info".getBytes(), "gender".getBytes(),
                    CompareFilter.CompareOp.EQUAL, binaryComparator);


            //增加多个过滤器
            FilterList filterList = new FilterList();
            filterList.addFilter(rowFilter);
            filterList.addFilter(valueFilter);


            Scan scan = new Scan();
            //增过滤器   多个过滤器
            scan.setFilter(filterList);


            ResultScanner scanner = student.getScanner(scan);
            Result result;
            while ((result = scanner.next()) != null) {
                print(result);
            }


        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    public static void print(Result result) {
        //取出rowkey
        String id = Bytes.toString(result.getRow());


        // 1、直接通过列名取值   如果列名不存在 返回对应类型的默认值
        byte[] bytes = result.getValue("info".getBytes(), "name".getBytes());
        //通过String的构造函数 构建字符串
        //String name = new String(bytes, "utf-8");

        //hbase  推荐方式
        String name = Bytes.toString(bytes);

        //int  类型  Bytes.toInt  转换
        int age = Bytes.toInt(result.getValue("info".getBytes(), "age".getBytes()));

        String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
        String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));

        System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz);
    }


}
